package tjs.study.notes.flink.kafka;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * Desc:
 * weixi: zhisheng_tian
 * blog: http://www.54tianzhisheng.cn/
 */
public class KafkaFlinkConsumerTest {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //key 反序列化
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest"); // latest earliest

        DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer<String>(
                "student",  //kafka topic
                new SimpleStringSchema(),  // String 序列化
                props))
                .setParallelism(1);


        DataStream<Student> studentDataStreamSource = dataStreamSource
                .map(new MapFunction<String, Student>() {
                    @Override
                    public Student map(String value) throws Exception {
                        Student student = JSONObject.parseObject(value, Student.class);
                        if (student.getAge() % 8 == 0) {
                            student.setNewAcount(student.getAcount() + 10);
                        }
                        return student;
                    }
                });
        studentDataStreamSource.print();
        // 存入mysql
//        studentDataStreamSource.addSink(new SinkToMySQL()).setParallelism(1);

        env.execute("Flink add data source");
    }

    public class MyTimestampsAndWatermarks implements AssignerWithPeriodicWatermarks<Student> {
        // 定义1分钟的容忍间隔时间，即允许数据的最大乱序时间
        private long maxOutofOrderness = 60 * 1000;
        // 观察到的最大时间戳
        private long currentMaxTs = Long.MIN_VALUE;

        @Override
        public Watermark getCurrentWatermark() {
            // 生成具有1分钟容忍度的水位线
            return new Watermark(currentMaxTs - maxOutofOrderness);
        }

        @Override
        public long extractTimestamp(Student element, long previousElementTimestamp) {
            //获取当前记录的时间戳
            long currentTs = element.timestamp;
            // 更新最大的时间戳
            currentMaxTs = Math.max(currentMaxTs, currentTs);
            // 返回记录的时间戳
            return currentTs;
        }
    }
}